package com.heiye.blog.ai.rpc;

import com.heiye.blog.ai.enums.ResponseCodeEnum;
import com.heiye.blog.framework.common.exception.BizException;
import com.heiye.blog.framework.common.response.Response;
import com.heiye.blog.oss.api.api.FileApi;
import com.heiye.blog.oss.api.model.dto.AudioStreamUploadDTO;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.retry.Retry;

import java.io.InputStream;
import java.time.Duration;
import java.util.Objects;

/**
 * @author: heiye
 * @date: 2025/11/21 下午4:03
 * @version: v1.0.0
 * @description: OSS RPC 服务
 */
@Slf4j
@Component
public class OSSRpcService {

    @Resource
    private FileApi fileApi;

    /**
     * 音频文件流式上传
     *
     * @param
     * @return
     */
    @SneakyThrows
    public Mono<String> audioStreamUpload(Flux<DataBuffer> audioStream, String audioType) {

        return DataBufferUtils.join(audioStream)
                // TTS 流异常保护
                .timeout(Duration.ofSeconds(30))
                .flatMap(dataBuffer ->
                        // 使用 Mono 创建一个资源，并声明如何使用这个资源
                        Mono.using(
                                // 资源构造
                                () -> dataBuffer.asInputStream(true),
                                // 逻辑处理 声明该操作是阻塞操作，并调用 subscribeOn
                                inputStream -> Mono.fromCallable(() -> {
                                    // 创建上传请求 DTO
                                    AudioStreamUploadDTO audioStreamUploadDTO = AudioStreamUploadDTO.builder()
                                            .inputStream(inputStream)
                                            .audioType(audioType)
                                            .build();

                                    // 调用 fileApi 执行上传
                                    Response<String> response = fileApi.audioStreamUpload(audioStreamUploadDTO);

                                    // 未上传成功则抛出业务异常
                                    if (Objects.isNull(response) || !response.isSuccess()) {
                                        throw new BizException(ResponseCodeEnum.AUDIO_UPLOAD_FAILED);
                                    }
                                    return response.getData();
                                }),
                                inputStream -> {
                                    // 自动关闭
                                    try {
                                        inputStream.close();
                                    } catch (Exception e) {
                                        log.info("==> 关闭输入流异常: {}", e.getMessage());
                                    }
                                    DataBufferUtils.release(dataBuffer);
                                }
                        ).subscribeOn(Schedulers.boundedElastic()))
                // 添加重试机制
                .retryWhen(Retry.backoff(3, Duration.ofMillis(200)));
    }
}
